package cn.yangliu.task.handler;

import cn.yangliu.task.TaskFactoryBean;
import cn.yangliu.task.entity.Task;
import cn.yangliu.task.entity.TaskVo;
import cn.yangliu.task.utils.TransformDataUtils;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.util.Objects;

/**
 * @author 杨柳
 * @date 2019-04-09
 */
public class TaskSubscription extends MessageListenerAdapter {

    private <T> T byteToObject(byte[] bytes) {
        Object obj = null;
        try (ByteArrayInputStream bi = new ByteArrayInputStream(bytes);
             ObjectInputStream oi = new ObjectInputStream(bi)) {
            obj = oi.readObject();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return (T) obj;
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {

        byte[] bytes = message.getChannel();
        String channel = new String(bytes);

        byte[] body = message.getBody();
        TaskVo taskVo = byteToObject(body);

        Task task = TransformDataUtils.getTransformData(taskVo);
        if (Objects.equals(channel, TaskPublisher.PUBLISH_TYPE_CANCLE)) {
            cancleTask(task);
        } else if (Objects.equals(channel, TaskPublisher.PUBLISH_TYPE_CREATE)) {
            createTask(task);
        } else if (Objects.equals(channel, TaskPublisher.PUBLISH_TYPE_RESTART)) {
            restartTask(task);
        } else {
            throw new UnsupportedOperationException();
        }
    }


    protected void cancleTask(Task task) {
        TaskFactoryBean.cancleTask(task.getId());
    }

    protected void createTask(Task task) {
        TaskFactoryBean.createTask(task);
    }

    protected void restartTask(Task task) {
        TaskFactoryBean.restartTask(task);
    }
}
